package com.ww.exercise.channel.actualcombat.producerconsumer

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicLong
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

/**
 * 极限压测工具类
 */
class StressTester {

    companion object {
        private val totalProduced = AtomicLong(0)
        private val totalConsumed = AtomicLong(0)
        private val totalErrors = AtomicLong(0)
    }

    /**
     * 场景1: 突发流量压测 - 瞬间大量数据涌入
     */
    suspend fun testBurstTraffic(): StressTestResult {
        println("=== 场景1: 突发流量压测 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        val processor = ChannelProcessor<String>(
            config = Config(
                channelCapacity = 1000, // 中等容量
                consumerCount = 4,
                produceTimeout = 100.milliseconds, // 短超时
                shutdownTimeout = 60.seconds
            )
        )

        // 慢消费者 - 模拟处理瓶颈
        processor.startConsumers { item ->
            delay(10) // 每个项目处理10ms
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()

        // 突发生产：100个协程同时生产
        val producerJobs = List(100) { producerId ->
            GlobalScope.launch {
                repeat(100) { i -> // 每个生产者生产100个项目
                    try {
                        val success = processor.produceWithTimeout("Burst-$producerId-$i")
                        if (success) {
                            totalProduced.incrementAndGet()
                        } else {
                            totalErrors.incrementAndGet()
                        }
                    } catch (e: Exception) {
                        totalErrors.incrementAndGet()
                    }
                }
            }
        }

        // 等待所有生产者完成
        producerJobs.forEach { it.join() }

        println("生产完成，等待消费...")
        delay(5000) // 给消费者时间处理

        // 优雅关闭
        val shutdownResult = processor.shutdown(30.seconds)

        val duration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "突发流量压测",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = duration,
            shutdownResult = shutdownResult
        )
    }

    /**
     * 场景2: 持续高负载压测 - 长时间高吞吐量
     */
    suspend fun testSustainedHighLoad(): StressTestResult {
        println("=== 场景2: 持续高负载压测 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        val processor = ChannelProcessor<String>(
            config = Config(
                channelCapacity = 500, // 较小容量，更容易触发背压
                consumerCount = 8,     // 更多消费者
                shutdownTimeout = 60.seconds
            )
        )

        // 中等速度消费者
        processor.startConsumers { item ->
            delay(5) // 每个项目处理5ms
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()
        val testDuration = 30.seconds // 测试持续30秒

        // 持续生产协程
        val producerJob = GlobalScope.launch {
            var counter = 0
            while (System.currentTimeMillis() - startTime < testDuration.inWholeMilliseconds) {
                repeat(100) { // 每次批量生产100个
                    val success = processor.tryProduce("Sustained-${counter++}")
                    if (success) {
                        totalProduced.incrementAndGet()
                    } else {
                        totalErrors.incrementAndGet()
                        delay(1) // 背压时短暂等待
                    }
                }
                delay(10) // 控制生产节奏
            }
        }

        // 监控协程 - 定期输出状态
        val monitorJob = GlobalScope.launch {
            while (isActive) {
                delay(5000)
                val status = processor.getBackpressureStatus()
                val metrics = processor.getMetrics()
                println("[Monitor] Queue: ${status.queueSize}/${status.capacity} (${"%.1f".format(status.utilization * 100)}%) " +
                        "Produced: ${metrics.produced}, Consumed: ${metrics.consumed}, " +
                        "Backpressure: ${if (status.hasBackpressure) "YES" else "NO"}")

                if (System.currentTimeMillis() - startTime > testDuration.inWholeMilliseconds) {
                    break
                }
            }
        }

        producerJob.join()
        monitorJob.cancel()

        println("测试时间结束，开始关闭...")
        val shutdownResult = processor.shutdown(30.seconds)

        val duration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "持续高负载压测",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = duration,
            shutdownResult = shutdownResult
        )
    }

    /**
     * 场景3: 极慢消费者压测 - 模拟严重消费瓶颈
     */
    suspend fun testSlowConsumer(): StressTestResult {
        println("=== 场景3: 极慢消费者压测 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        val processor = ChannelProcessor<String>(
            config = Config(
                channelCapacity = 100, // 小容量，快速触发背压
                consumerCount = 2,
                produceTimeout = 50.milliseconds, // 很短超时
                shutdownTimeout = 60.seconds
            )
        )

        // 极慢消费者 - 模拟严重处理瓶颈
        processor.startConsumers { item ->
            delay(100) // 每个项目处理100ms - 很慢！
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()

        // 快速生产者
        val producerJob = GlobalScope.launch {
            repeat(1000) { i -> // 生产1000个项目
                val success = processor.produceWithTimeout("Slow-$i")
                if (success) {
                    totalProduced.incrementAndGet()
                } else {
                    totalErrors.incrementAndGet()
                    if (i % 100 == 0) {
                        val status = processor.getBackpressureStatus()
                        println("背压事件 $i: 队列 ${status.queueSize}/${status.capacity}")
                    }
                }
                delay(1) // 很快的生产速度
            }
        }

        producerJob.join()
        delay(2000) // 给消费者一些时间

        val shutdownResult = processor.shutdown(30.seconds)

        val duration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "极慢消费者压测",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = duration,
            shutdownResult = shutdownResult
        )
    }

    /**
     * 场景4: 内存极限压测 - 测试内存使用和GC表现
     */
    suspend fun testMemoryStress(): StressTestResult {
        println("=== 场景4: 内存极限压测 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        // 使用大对象测试内存
        data class LargeObject(val id: Int, val data: String = "X".repeat(1024)) // 1KB对象

        val processor = ChannelProcessor<LargeObject>(
            config = Config(
                channelCapacity = 10000, // 大容量，占用更多内存
                consumerCount = 4,
                shutdownTimeout = 60.seconds
            )
        )

        // 快速消费者
        processor.startConsumers { item ->
            // 简单处理
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()

        // 内存密集型生产
        val producerJob = GlobalScope.launch {
            repeat(50000) { i -> // 生产5万个较大对象
                val success = processor.produce(LargeObject(i))
                if (success) {
                    totalProduced.incrementAndGet()
                } else {
                    totalErrors.incrementAndGet()
                }

                // 每生产1000个输出内存状态
                if (i % 1000 == 0) {
                    val runtime = Runtime.getRuntime()
                    val usedMemory = (runtime.totalMemory() - runtime.freeMemory()) / (1024 * 1024)
                    println("内存使用: ${usedMemory}MB, 生产进度: $i/50000")
                }
            }
        }

        producerJob.join()
        val shutdownResult = processor.shutdown(30.seconds)

        val duration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "内存极限压测",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = duration,
            shutdownResult = shutdownResult
        )
    }

    /**
     * 场景5: 混合负载压测 - 模拟真实场景的波动负载
     */
    suspend fun testMixedWorkload(): StressTestResult {
        println("=== 场景5: 混合负载压测 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        val processor = ChannelProcessor<String>(
            config = Config(
                channelCapacity = 2000,
                consumerCount = 6,
                shutdownTimeout = 60.seconds
            )
        )

        // 混合速度消费者 - 模拟不同处理能力的消费者
        processor.startConsumers { consumerId, item ->
            // 模拟不同处理时间：20%快速，60%中等，20%慢速
            val processingTime = when (consumerId % 5) {
                0 -> 2L // 快速消费者
                1, 2, 3 -> 10L // 中等消费者
                else -> 50L // 慢速消费者
            }
            delay(processingTime)
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()
        val testDuration = 45.seconds

        // 波动生产者 - 模拟真实负载波动
        val producerJob = GlobalScope.launch {
            var phase = 0
            while (System.currentTimeMillis() - startTime < testDuration.inWholeMilliseconds) {
                val elapsed = System.currentTimeMillis() - startTime
                phase = (elapsed / 10000).toInt() // 每10秒一个阶段

                val productionRate = when (phase % 4) {
                    0 -> 10 // 低负载
                    1 -> 100 // 中等负载
                    2 -> 500 // 高负载
                    else -> 50 // 恢复期
                }

                repeat(productionRate) {
                    val success = processor.tryProduce("Mixed-Phase$phase-$it")
                    if (success) {
                        totalProduced.incrementAndGet()
                    } else {
                        totalErrors.incrementAndGet()
                    }
                }

                delay(1000) // 每秒一次生产周期

                // 输出阶段信息
                if (elapsed % 10000 < 1000) { // 每阶段开始输出
                    val status = processor.getBackpressureStatus()
                    println("阶段 $phase: 生产速率 $productionRate/秒, 队列 ${status.queueSize}/${status.capacity}")
                }
            }
        }

        producerJob.join()
        val shutdownResult = processor.shutdown(30.seconds)

        val duration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "混合负载压测",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = duration,
            shutdownResult = shutdownResult
        )
    }

    /**
     * 场景6: 关闭压力测试 - 在高压下测试优雅关闭
     */
    suspend fun testShutdownUnderLoad(): StressTestResult {
        println("=== 场景6: 关闭压力测试 ===")
        totalProduced.set(0)
        totalConsumed.set(0)
        totalErrors.set(0)

        val processor = ChannelProcessor<String>(
            config = Config(
                channelCapacity = 500,
                consumerCount = 4,
                shutdownTimeout = 10.seconds // 较短关闭超时
            )
        )

        // 慢消费者
        processor.startConsumers { item ->
            delay(20) // 每个项目处理20ms
            totalConsumed.incrementAndGet()
        }

        val startTime = System.currentTimeMillis()

        // 持续生产，同时在后台准备关闭
        val producerJob = GlobalScope.launch {
            var counter = 0
            while (isActive) {
                repeat(50) {
                    processor.produce("ShutdownTest-${counter++}")
                    totalProduced.incrementAndGet()
                }
                delay(100)
            }
        }

        // 在负载下启动关闭
        delay(3000) // 让系统先运行3秒

        println("在负载下启动关闭...")
        val shutdownStart = System.currentTimeMillis()
        val shutdownResult = processor.shutdown(5.seconds) // 很短的关闭超时

        val shutdownDuration = System.currentTimeMillis() - shutdownStart
        producerJob.cancel()

        val totalDuration = System.currentTimeMillis() - startTime
        return StressTestResult(
            scenario = "关闭压力测试",
            produced = totalProduced.get(),
            consumed = totalConsumed.get(),
            errors = totalErrors.get(),
            duration = totalDuration,
            shutdownResult = shutdownResult,
            additionalInfo = "关闭耗时: ${shutdownDuration}ms"
        )
    }
}

/**
 * 压测结果
 */
data class StressTestResult(
    val scenario: String,
    val produced: Long,
    val consumed: Long,
    val errors: Long,
    val duration: Long,
    val shutdownResult: ShutdownResult,
    val additionalInfo: String = ""
) {
    val throughput: Double get() = consumed.toDouble() / duration * 1000
    val errorRate: Double get() = if (produced > 0) errors.toDouble() / produced else 0.0
    val successRate: Double get() = if (produced > 0) consumed.toDouble() / produced else 1.0

    fun printReport() {
        println("\n" + "=".repeat(60))
        println("压测报告: $scenario")
        println("=".repeat(60))
        println("运行时间: ${duration}ms")
        println("生产数量: $produced")
        println("消费数量: $consumed")
        println("错误数量: $errors")
        println("吞吐量: ${"%.2f".format(throughput)} 项目/秒")
        println("成功率: ${"%.2f".format(successRate * 100)}%")
        println("错误率: ${"%.2f".format(errorRate * 100)}%")
        println("关闭结果: ${shutdownResult.message}")
        if (additionalInfo.isNotEmpty()) {
            println("附加信息: $additionalInfo")
        }
        println("=".repeat(60))
    }
}

/**
 * 扩展函数：支持按消费者ID区分的处理器
 */
fun <T> ChannelProcessor<T>.startConsumers(
    processor: suspend (Int, T) -> Unit
) {
    repeat(config.consumerCount) { consumerId ->
        scope.launch {
            for (item in channel) {
                processItem(consumerId, item) { itemToProcess ->
                    processor(consumerId, itemToProcess)
                }
            }
        }
    }
}

// 修改processItem以支持带消费者ID的处理
private suspend fun <T> ChannelProcessor<T>.processItem(
    consumerId: Int,
    item: T,
    processor: suspend (T) -> Unit
) {
    try {
        processor(item)
        consumedCount.incrementAndGet()
    } catch (e: Exception) {
        println("Consumer $consumerId error processing item: ${e.message}")
    } finally {
//        queueSize.decrementAndGet()
//        pendingInChannel?.decrementAndGet()
    }
}